package com.alan.mr.weibo;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

/**
 * Created by Alan on 2017/10/2.
 */
public class ThirdJob {

    public static void main(String[] args) {

        Configuration config = new Configuration();
        config.set("fs.defaults", "hdfs://192.168.208.106:8020");
        config.set("yarn.resourcemanager.hostname", "192.168.208.106");
        try {
            Job job = Job.getInstance(config);
            job.setJarByClass(ThirdJob.class);
            job.setJobName("third");
//            job.setInputFormatClass(KeyValueTextInputFormat.class);

            //把微博总数加载到内存
            job.addCacheFile(new Path("E:\\sxt\\1-MapReduce\\data\\weibo1\\part-r-00003").toUri());
            //把df加载到内存
            job.addCacheFile(new Path("E:\\sxt\\1-MapReduce\\data\\weibo2\\part-r-00000").toUri());

            job.setMapperClass(ThirdMapper.class);
            job.setReducerClass(ThirdReducer.class);

            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(Text.class);

            FileSystem fs = FileSystem.get(config);
            FileInputFormat.addInputPath(job, new Path("E:\\sxt\\1-MapReduce\\data\\weibo1"));
            Path path = new Path("E:\\sxt\\1-MapReduce\\data\\weibo3");
            if (fs.exists(path)) {
                fs.delete(path);
            }
            FileOutputFormat.setOutputPath(job, path);

            boolean waitForCompletion = job.waitForCompletion(true);
            if(waitForCompletion) {
                System.out.println("执行job成功");
            }
        }catch (Exception e) {
            e.printStackTrace();
        }
    }

}
